package com.hx.vending.mqtt;

import org.apache.commons.codec.binary.Base64;
import org.apache.log4j.Logger;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.TrustManagerFactory;
import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.security.KeyFactory;
import java.security.KeyStore;
import java.security.PrivateKey;
import java.security.SecureRandom;
import java.security.cert.CertificateFactory;
import java.security.cert.X509Certificate;
import java.security.interfaces.RSAPrivateKey;
import java.security.spec.PKCS8EncodedKeySpec;
import java.util.UUID;

public class ClientMQTT extends Thread {

	public static final String HOST = "ssl://47.105.178.86:9090";
//	private static final String HOST = "tcp://47.105.178.86:1883";

	private static MqttClient client;
	private MqttConnectOptions options;
	private String userName = "admin";//非必须
	private String passWord = "public";  //非必须


	public static String path = "";
	private static Logger logger = Logger.getLogger(ClientMQTT.class);


	/**
	 * 线程类
	 */
	@Override
	public void run() {
		String clientid = "admin"+UUID.randomUUID().toString().replace("-", "");
		System.out.println(String.format("--------------------------------主题线程开启--------------------------------"));
		System.out.println("clientid:"+clientid);
		logger.error(clientid);
		try {
			// host为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
			client = new MqttClient(HOST, clientid, new MemoryPersistence());
			// MQTT的连接设置
			options = new MqttConnectOptions();
			// 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，设置为true表示每次连接到服务器都以新的身份连接
			options.setCleanSession(false);
			// 设置连接的用户名
			options.setUserName(userName);
			// 设置连接的密码
			options.setPassword(passWord.toCharArray());
			// 设置超时时间 单位为秒
			options.setConnectionTimeout(10);
			// 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
			options.setKeepAliveInterval(20);
			//自动重连
			options.setAutomaticReconnect(true);
			//获取conf路径
			String path = this.getClass().getResource("").getPath();
//			System.out.println(path);
			String[] split = path.split("com");
			String str =  split[0].substring(0,  split[0].length());
			String sendPath = str + "conf/mqtt/";
//			System.out.println(sendPath);
			this.path = sendPath;
			//证书路径
			SSLSocketFactory sslSocketFactory = getSSLSocktet(sendPath + "ca.crt", sendPath + "client.crt",sendPath + "client.pem", "");
			options.setSocketFactory(sslSocketFactory);
			// 设置回调
			client.setCallback(new PushCallback());
			//setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息
			//遗嘱        options.setWill(topic, "close".getBytes(), 2, true);
			boolean boo = true;
			while(boo){
				try {
					client.connect(options);
					if (client.isConnected()) {
						boo = false;
						logger.error("连接到mqtt");
					}
				} catch (Exception e) {
					boo = true;
					logger.error("------------------------订阅主题连接失败，10秒后重连-------------------------");
					Thread.sleep(10000);
				}
			}

			/**
			 * 订阅主题
			 * Connect             终端连接服务器
			 * Offline             下线遗嘱
			 * Heartbeat           心跳包
			 * DeliveryReceive     出货指令终端应答
			 * Alarm               故障上传
			 * Position            位置上传
			 * DeliverySuccess     出货成功
			 * SetUse              设备使用停用
			 * RefreshGoodsReceive 刷新货道商品信息
			 * ChangeBanner        刷新广告图片
			 * TestResult          货道测试结果
			 * DetectConnections   检测mqtt
			 * TestMouthReceive    货道测试应答
			 * AdjustAudioSuccess  调整终端音量应答
			 * UploadShotReceive   设备屏幕截图应答
			 */
			String[] a=new String[]{"Connect","Offline","Heartbeat","DeliveryReceive","Alarm","Position","DeliverySuccess", "SetUse", "RefreshGoodsReceive",
					"ChangeBanner", "TestResult", "DetectConnections", "TestMouthReceive", "AdjustAudioSuccess", "UploadShotReceive","UploadPhotoReceive",
					"VibrateAlarm"};
			int qos=0;
			for (String string : a) {
				client.subscribe(string,qos);
			}
			System.out.println(String.format("--------------------------------主题订阅完成--------------------------------"));
		} catch (Exception e) {
			System.out.println(e.getMessage());
			//e.printStackTrace();
		}
	}


	/**
	 * MQTT连接对象,用于发送消息
	 * @return
	 */
	public static MqttClient getClient() {
		return client;
	}

	public static void setClient(MqttClient client) {
		ClientMQTT.client = client;
	}


	//mqtt证书获取
	public SSLSocketFactory getSSLSocktet(String caPath, String crtPath, String keyPath, String password) throws Exception {
		// CA certificate is used to authenticate server
		CertificateFactory cAf = CertificateFactory.getInstance("X.509");
		FileInputStream caIn = new FileInputStream(caPath);
		X509Certificate ca = (X509Certificate) cAf.generateCertificate(caIn);
		KeyStore caKs = KeyStore.getInstance("JKS");
		caKs.load(null, null);
		caKs.setCertificateEntry("ca-certificate", ca);
		TrustManagerFactory tmf = TrustManagerFactory.getInstance("PKIX");
		tmf.init(caKs);

		CertificateFactory cf = CertificateFactory.getInstance("X.509");
		FileInputStream crtIn = new FileInputStream(crtPath);
		X509Certificate caCert = (X509Certificate) cf.generateCertificate(crtIn);

		crtIn.close();
		// client key and certificates are sent to server so it can authenticate
		// us
		KeyStore ks = KeyStore.getInstance(KeyStore.getDefaultType());
//      ks.load(caIn,password.toCharArray());
		ks.load(null, null);
		ks.setCertificateEntry("certificate", caCert);
		ks.setKeyEntry("private-key", getPrivateKey(keyPath), password.toCharArray(),
				new java.security.cert.Certificate[]{caCert}  );
		KeyManagerFactory kmf = KeyManagerFactory.getInstance("PKIX");
		kmf.init(ks, password.toCharArray());
//      keyIn.close();

		// finally, create SSL socket factory
		SSLContext context = SSLContext.getInstance("TLSv1");

		context.init(kmf.getKeyManagers(),tmf.getTrustManagers(), new SecureRandom());

		return context.getSocketFactory();
	}

	public PrivateKey getPrivateKey(String path) throws Exception{
		org.apache.commons.codec.binary.Base64 base64=new Base64();
		byte[] buffer=   base64.decode(getPem(path));

		PKCS8EncodedKeySpec keySpec= new PKCS8EncodedKeySpec(buffer);
		KeyFactory keyFactory= KeyFactory.getInstance("RSA");
		return (RSAPrivateKey) keyFactory.generatePrivate(keySpec);
	}

	private String getPem(String path) throws Exception{
		FileInputStream fin=new FileInputStream(path);
		BufferedReader br= new BufferedReader(new InputStreamReader(fin));
		String readLine= null;
		StringBuilder sb= new StringBuilder();
		while((readLine= br.readLine())!=null){
			if(readLine.charAt(0)=='-'){
				continue;
			}else{
				sb.append(readLine);
				sb.append('\r');
			}
		}
		fin.close();
		return sb.toString();
	}

}